package com.shujia.core

import com.shujia.core.Demo10Join.Student
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo23Broadcast {
  def main(args: Array[String]): Unit = {
    // 假设 有一份 黑名单 包含了 多个 学生ID

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo23Broadcast")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 学生黑名单RDD
    val blackListRDD: RDD[String] = sc.parallelize(List("1500100033", "1500100133", "1500100233", "1500100333"))


    // 读取学生数据及分数数据 并将每一行数据转换成样例类对象
    val stuRDD: RDD[Student] = sc
      .textFile("Spark/data/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        val id: String = splits(0)
        val name: String = splits(1)
        val age: Int = splits(2).toInt
        val gender: String = splits(3)
        val clazz: String = splits(4)
        Student(id, name, age, gender, clazz)
      })

    // 将RDD变为Set
    val blackListSet: Set[String] = blackListRDD.collect().toSet

    // 处理students数据 并基于黑名单 找出黑名单中学生的信息
    stuRDD

      /**
       * 算子内部的代码会以Task的形式发送到Executor中执行
       * 在算子内部 用到了一个Scala中的集合：blackListSet
       * 该集合是在算子外部定义的 即Driver端
       * 任务在执行时 会将该Set以副本形式Copy到每个Task中
       * 假设数据有10个分区->10个Task->Set会被Copy10次
       * 又因为在一个Executor中可以执行多个Task
       * 所以没有必要每个Task上去封装一份Set 可以将Set Copy到每个Executor中
       * 那么在Executor上执行的Task就可以直接从Executor拉去Set数据
       * 避免多次传输造成的网络开销
       * 分区的数量>>Executor的数量
       *
       */
      .filter(stu => blackListSet.contains(stu.id))
      .foreach(println)

    // 使用广播变量避免多次Copy数据
    // 使用SparkContext进行广播变量至每个Executor中
    val broSet: Broadcast[Set[String]] = sc.broadcast(blackListSet)
    stuRDD
      .filter(stu => {
        // 从Executor中获取广播的Set
        // 第一个Task在执行时，如果Executor中没有
        // 则会去向Driver端申请并拉取到Executor中并保存起来
        // 后续的Task就可以直接使用
        val broBlackListSet: Set[String] = broSet.value
        broBlackListSet.contains(stu.id)
      }).foreach(println)


  }

}
